package com.nulldev.util.io.legacy.cstreams;

import java.io.IOException;
import java.io.InputStream;

import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutorManager;

public class CoderInputStream extends InputStream {
	private final CoderThread ct;

	private volatile InputStream in;

	protected CoderInputStream(final InputStream in, final Coder coder) throws IOException {
		this.ct = new CoderThread(coder, in);
		this.in = this.ct.getInputStreamSink();
		ExecutorManager.get().queueTask(this.ct);
	}

	@Override
	public int read() throws IOException {
		return in.read();
	}

	@Override
	public int read(byte b[]) throws IOException {
		return read(b, 0, b.length);
	}

	@Override
	public int read(byte b[], int off, int len) throws IOException {
		return in.read(b, off, len);
	}

	@Override
	public long skip(long n) throws IOException {
		return in.skip(n);
	}

	@Override
	public int available() throws IOException {
		return in.available();
	}

	@Override
	public void close() throws IOException {
		in.close();

		try {
			ct.join();
		} catch (InterruptedException e) {
			throw new IOException(e);
		}

		ct.checkForException();
	}

	@Override
	public synchronized void mark(int readlimit) {
		in.mark(readlimit);
	}

	@Override
	public synchronized void reset() throws IOException {
		in.reset();
	}

	@Override
	public boolean markSupported() {
		return in.markSupported();
	}
}
